<!DOCTYPE html>
<html>

  <head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
	<meta http-equiv="Cache-Control" content="no-siteapp" />

    <title>自定义流的实现和使用</title>
    <meta name="description" content="tbox中提供了常用的一些stream模块，例如：data、file、http、sock等，可以通过指定不同的url，使用相同的接口进行数据流的读写，非常的方便。例如：    // init stream    tb_stream_ref_t stream = tb_stream_init_from_url("h...">

    
    <meta name="keywords" content="tbox,stream,流,tboox" /> 

    <!-- qq oauth -->
    <meta property="qc:admins" content="5211601217706727767255" />

    <!--icon -->
    <link rel="shortcut icon" href="/favicon.ico" type="image/x-icon">
    <link rel="icon" sizes="192x192" href="/static/img/nice-highres.png" />
	<link rel="apple-touch-icon-precomposed" href="/static/img/apple-touch-icon-57x57-precomposed.png" />
	<link rel="apple-touch-icon-precomposed" sizes="72x72" href="/static/img/apple-touch-icon-72x72-precomposed.png" />
	<link rel="apple-touch-icon-precomposed" sizes="114x114" href="/static/img/apple-touch-icon-114x114-precomposed.png" />
	<link rel="apple-touch-icon-precomposed" sizes="144x144" href="/static/img/apple-touch-icon-144x144-precomposed.png" />
	<link rel="apple-touch-icon-precomposed" sizes="180x180" href="/static/img/retinahd_icon.png" />
	<meta name="msapplication-TileImage" content="/static/img/retinahd_icon.png" />
	
    <link rel="stylesheet" href=" /css/fontawesome/css/font-awesome.min.css ">
    <link rel="stylesheet" href=" /css/main.css ">
    <link rel="canonical" href="https://tboox.org/cn/2016/03/17/stream-custom/">
    <link rel="alternate" type="application/rss+xml" title="TBOOX Open Source Project" href="https://tboox.org/feed.xml ">
    <link rel="alternate" hreflang="en" href="https://tboox.org/" />
    <link rel="alternate" hreflang="zh-Hans" href="https://tboox.org/cn/" />

    <!-- css -->
    <link href="/css/reward.css" rel="stylesheet" type="text/css"> 




    <script type="text/javascript">
    function isPC(){    
        var userAgentInfo = navigator.userAgent;  
        var Agents = new Array("Android", "iPhone", "SymbianOS", "Windows Phone", "iPad", "iPod");    
        var flag = true;    
        for (var v = 0; v < Agents.length; v++) {    
            if (userAgentInfo.indexOf(Agents[v]) > 0) { flag = false; break; }    
        }    
        return flag;    
    }
    </script>

<!-- baidu ads -->



    <!-- baidu ads -->

</head>


  <body>

    <header id="top">
    <div class="wrapper">
        <a href="/cn" class="brand">TBOOX</a>
        <button id="headerMenu" class="menu"><i class="fa fa-bars"></i></button>
        <nav id="headerNav">
            <ul>
                <li>
                    
                    <a href="/?lang=0">
                    
                        <i class="fa fa-home"></i>English
                    </a>
                </li>

                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/project/">
                            
                        
                            <i class="fa fa-bookmark"></i>项目
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/archive/">
                            
                        
                            <i class="fa fa-archive"></i>归档
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/category/">
                            
                        
                            <i class="fa fa-th-list"></i>分类
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/tag/">
                            
                        
                            <i class="fa fa-tags"></i>标记
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/docs/">
                            
                        
                            <i class="fa fa-book"></i>文档
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="https://xmake.io/#/zh-cn/about/contact" target="_blank" >
                            
                        
                            <i class="fa fa-forumbee"></i>社区
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/donation/">
                            
                        
                            <i class="fa fa-heart"></i>捐助
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                    
                    <li>
                        
                            
                            <a href="/cn/about/">
                            
                        
                            <i class="fa fa-user"></i>关于
                        </a>
                    </li>
                    
                    
                
                    
                    
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                    
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
                    
                
            </ul>
        </nav>
    </div>
</header>



        <div class="page clearfix" post>
    <div class="left">
        <h1>自定义流的实现和使用</h1>
        <div class="label">

            <div class="label-card">
                <i class="fa fa-calendar"></i>2016-03-17
            </div>

            <div class="label-card">
                
            </div>

            <div class="label-card">
                
            </div>

            <div class="label-card">
            


<!-- <span class="point">•</span> -->
<span class="categories">
  <i class="fa fa-th-list"></i>
  
    
        <a href="/category/#tbox" title="Category: tbox" rel="category">tbox</a>
    
  

  <!-- <span class="point">•</span> -->
</span>


            </div>

            <div class="label-card">
            
<!-- <span class="point">•</span> -->
<span class="pageTag">
  <i class="fa fa-tags"></i>
  
    
        <!--a href="/tag/#tbox" title="Tag: tbox" rel="tag">tbox</a-->
        <a href="/cn/tag/#tbox" title="Tag: tbox" rel="tag">tbox</a>&nbsp;
    
        <!--a href="/tag/#stream" title="Tag: stream" rel="tag">stream</a-->
        <a href="/cn/tag/#stream" title="Tag: stream" rel="tag">stream</a>&nbsp;
    
        <!--a href="/tag/#%E6%B5%81" title="Tag: 流" rel="tag">流</a-->
        <a href="/cn/tag/#流" title="Tag: 流" rel="tag">流</a>
    
  

</span>

            </div>

        </div>
        <hr>
        <article itemscope itemtype="http://schema.org/BlogPosting">
        <p>tbox中提供了常用的一些stream模块，例如：data、file、http、sock等，可以通过指定不同的url，使用相同的接口
进行数据流的读写，非常的方便。</p>

<p>例如：</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// init stream
</span>    <span class="n">tb_stream_ref_t</span> <span class="n">stream</span> <span class="o">=</span> <span class="n">tb_stream_init_from_url</span><span class="p">(</span><span class="s">"http://www.xxx.com/file.txt"</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">stream</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// open stream
</span>        <span class="k">if</span> <span class="p">(</span><span class="n">tb_stream_open</span><span class="p">(</span><span class="n">stream</span><span class="p">))</span>
        <span class="p">{</span>
            <span class="c1">// read line
</span>            <span class="n">tb_long_t</span> <span class="n">size</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
            <span class="n">tb_char_t</span> <span class="n">line</span><span class="p">[</span><span class="n">TB_STREAM_BLOCK_MAXN</span><span class="p">];</span>
            <span class="k">while</span> <span class="p">((</span><span class="n">size</span> <span class="o">=</span> <span class="n">tb_stream_bread_line</span><span class="p">(</span><span class="n">stream</span><span class="p">,</span> <span class="n">line</span><span class="p">,</span> <span class="k">sizeof</span><span class="p">(</span><span class="n">line</span><span class="p">)))</span> <span class="o">&gt;=</span> <span class="mi">0</span><span class="p">)</span>
            <span class="p">{</span>
                <span class="c1">// trace
</span>                <span class="n">tb_trace_i</span><span class="p">(</span><span class="s">"line: %s"</span><span class="p">,</span> <span class="n">line</span><span class="p">);</span>
            <span class="p">}</span>
        <span class="p">}</span>

        <span class="c1">// exit stream
</span>        <span class="n">tb_stream_exit</span><span class="p">(</span><span class="n">stream</span><span class="p">);</span>
    <span class="p">}</span>
</code></pre>
</div>

<p>这样的好处是，操作io的模块不需要关心实际的数据流协议，只管从stream中读写数据就行了，实现数据和业务逻辑的解耦。。</p>

<p>但是tbox提供的这些内置stream模块，有时候没法完全咱们的实际需求，例如：</p>

<p>我想读取一个实时数据流的缓存队列，这个数据流一段会不停的送入数据进来，另外一段会不停的读取数据，如果数据不够，就会进入等待</p>

<p>这其实是个很有用的功能，我的很多需求都会用到，例如：流媒体的一些实时数据获取和复用等等。。</p>

<p>那如何实现这样一个stream模块，让tbox的stream接口支持呢，我们只要实现一个自定义的流模块就好，实现起来也不复杂</p>

<p>我们先定义个一个stream类型，例如：</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// 用户自定义流类型：实时流
</span>    <span class="cp">#define TB_STREAM_TYPE_REAL         (TB_STREAM_TYPE_USER + 1)
</span>
    <span class="c1">// 定义一个控制流代码，之后tb_stream_ctrl需要
</span>    <span class="cp">#define TM_STREAM_CTRL_REAL_PUSH    TB_STREAM_CTRL(TM_STREAM_TYPE_REAL, 1)
</span></code></pre>
</div>

<p>定义个自定义流的数据结构，用于维护咱们的私有数据</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// 实时流类型
</span>    <span class="k">typedef</span> <span class="k">struct</span> <span class="n">__tb_stream_real_t</span>
    <span class="p">{</span>
        <span class="c1">// 这里定义了一个数据块buffer的队列，用于缓存不断送入的数据
</span>        <span class="n">tb_queue_ref_t</span>      <span class="n">buffers</span><span class="p">;</span>

        <span class="c1">// 总的数据大小
</span>        <span class="n">tb_size_t</span>           <span class="n">size</span><span class="p">;</span>

    <span class="p">}</span><span class="n">tb_stream_real_t</span><span class="p">,</span> <span class="o">*</span><span class="n">tb_stream_real_ref_t</span><span class="p">;</span>

    <span class="c1">// 定义一个buffer块类型，用于维护单个数据块
</span>    <span class="k">typedef</span> <span class="k">struct</span> <span class="n">__tm_real_buffer_t</span>
    <span class="p">{</span>
        <span class="c1">// 数据地址
</span>        <span class="n">tb_byte_t</span><span class="o">*</span>          <span class="n">data</span><span class="p">;</span>

        <span class="c1">// 这个buffer总大小
</span>        <span class="n">tb_size_t</span>           <span class="n">size</span><span class="p">;</span>

        <span class="c1">// 在这个buffer中，当前读取到的数据
</span>        <span class="n">tb_size_t</span>           <span class="n">read</span><span class="p">;</span>

    <span class="p">}</span><span class="n">tm_real_buffer_t</span><span class="p">,</span> <span class="o">*</span><span class="n">tm_real_buffer_ref_t</span><span class="p">;</span>
</code></pre>
</div>

<p>创建一个stream实例，注册一些需要的回调接口</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// 初始化创建个一个实时流
</span>    <span class="n">tb_stream_ref_t</span> <span class="nf">tb_stream_init_real</span><span class="p">()</span>
    <span class="p">{</span>
        <span class="k">return</span> <span class="n">tb_stream_init</span><span class="p">(</span>  <span class="n">TB_STREAM_TYPE_REAL</span>
                            <span class="p">,</span>   <span class="k">sizeof</span><span class="p">(</span><span class="n">tb_stream_real_t</span><span class="p">)</span>
                            <span class="p">,</span>   <span class="mi">0</span>           <span class="c1">// stream缓存大小（file/sock有用），这里禁用了，因为咱们的流不需要缓存读取
</span>                            <span class="p">,</span>   <span class="n">tb_stream_real_open</span>
                            <span class="p">,</span>   <span class="n">tb_stream_real_clos</span>
                            <span class="p">,</span>   <span class="n">tb_stream_real_exit</span>
                            <span class="p">,</span>   <span class="n">tb_stream_real_ctrl</span>
                            <span class="p">,</span>   <span class="n">tb_stream_real_wait</span>
                            <span class="p">,</span>   <span class="n">tb_stream_real_read</span>
                            <span class="p">,</span>   <span class="n">tb_null</span>     <span class="c1">// 写回调，这里不需要
</span>                            <span class="p">,</span>   <span class="n">tb_null</span>     <span class="c1">// seek，我们这里不需要
</span>                            <span class="p">,</span>   <span class="n">tb_null</span>     <span class="c1">// 刷新写数据，不需要
</span>                            <span class="p">,</span>   <span class="n">tb_null</span><span class="p">);</span>   <span class="c1">// kill当前的stream，很少用，一般用于中断内部读写
</span>    <span class="p">}</span>
</code></pre>
</div>

<p>下面就是具体的回调接口实现了</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// 实现open回调接口，用于打开stream，tb_stream_open会用到
</span>    <span class="k">static</span> <span class="n">tb_bool_t</span> <span class="nf">tb_stream_real_open</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return_val</span><span class="p">(</span><span class="n">rstream</span><span class="p">,</span> <span class="n">tb_false</span><span class="p">);</span>

        <span class="c1">// 初始化一个buffer队列，并注册自动释放接口：tb_real_buffer_exit，之后有说明
</span>        <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span> <span class="o">=</span> <span class="n">tb_queue_init</span><span class="p">(</span><span class="mi">0</span><span class="p">,</span> <span class="n">tb_element_mem</span><span class="p">(</span><span class="k">sizeof</span><span class="p">(</span><span class="n">tb_real_buffer_t</span><span class="p">),</span> <span class="n">tb_real_buffer_exit</span><span class="p">,</span> <span class="n">tb_null</span><span class="p">));</span>

        <span class="c1">// init size
</span>        <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">size</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>

        <span class="c1">// ok
</span>        <span class="k">return</span> <span class="o">!!</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="c1">// 实现close回调接口，用于关闭stream，tb_stream_clos会用到
</span>    <span class="k">static</span> <span class="n">tb_bool_t</span> <span class="nf">tb_stream_real_clos</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return_val</span><span class="p">(</span><span class="n">rstream</span><span class="p">,</span> <span class="n">tb_false</span><span class="p">);</span>
        
        <span class="c1">// exit buffers
</span>        <span class="k">if</span> <span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">)</span> <span class="n">tb_queue_exit</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">);</span>
        <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span> <span class="o">=</span> <span class="n">tb_null</span><span class="p">;</span>

        <span class="c1">// ok
</span>        <span class="k">return</span> <span class="n">tb_true</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="c1">// 实现exit回调接口，用于销毁stream，tb_stream_exit会用到
</span>    <span class="k">static</span> <span class="n">tb_void_t</span> <span class="nf">tb_stream_real_exit</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return</span><span class="p">(</span><span class="n">rstream</span><span class="p">);</span>
        
        <span class="c1">// exit buffers
</span>        <span class="k">if</span> <span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">)</span> <span class="n">tb_queue_exit</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">);</span>
        <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span> <span class="o">=</span> <span class="n">tb_null</span><span class="p">;</span>

        <span class="c1">// clear size
</span>        <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">size</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="c1">// 实现read回调接口，用于读取数据，tb_stream_read/tb_stream_bread等接口会用到
</span>    <span class="k">static</span> <span class="n">tb_long_t</span> <span class="nf">tb_stream_real_read</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">,</span> <span class="n">tb_byte_t</span><span class="o">*</span> <span class="n">data</span><span class="p">,</span> <span class="n">tb_size_t</span> <span class="n">size</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return_val</span><span class="p">(</span><span class="n">rstream</span> <span class="o">&amp;&amp;</span> <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">,</span> <span class="o">-</span><span class="mi">1</span><span class="p">);</span>

        <span class="c1">// check
</span>        <span class="n">tb_check_return_val</span><span class="p">(</span><span class="n">data</span><span class="p">,</span> <span class="o">-</span><span class="mi">1</span><span class="p">);</span>
        <span class="n">tb_check_return_val</span><span class="p">(</span><span class="n">size</span><span class="p">,</span> <span class="mi">0</span><span class="p">);</span>

        <span class="c1">// 依次从队列头部读取每块buffer的数据，直到读满为止
</span>        <span class="n">tb_long_t</span> <span class="n">read</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
        <span class="k">while</span> <span class="p">(</span><span class="n">read</span> <span class="o">&lt;</span> <span class="n">size</span> <span class="o">&amp;&amp;</span> <span class="n">tb_queue_size</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">))</span>
        <span class="p">{</span>
            <span class="c1">// get buffer
</span>            <span class="n">tb_real_buffer_ref_t</span> <span class="n">buffer</span> <span class="o">=</span> <span class="n">tb_queue_get</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">);</span>
            <span class="n">tb_assert_and_check_break</span><span class="p">(</span><span class="n">buffer</span> <span class="o">&amp;&amp;</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">data</span> <span class="o">&amp;&amp;</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">size</span><span class="p">);</span>

            <span class="c1">// read data
</span>            <span class="k">if</span> <span class="p">(</span><span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span> <span class="o">&lt;</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">size</span><span class="p">)</span>
            <span class="p">{</span>
                <span class="c1">// calculate the need size
</span>                <span class="n">tb_size_t</span> <span class="n">need</span> <span class="o">=</span> <span class="n">tb_min</span><span class="p">(</span><span class="n">size</span> <span class="o">-</span> <span class="n">read</span><span class="p">,</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">size</span> <span class="o">-</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span><span class="p">);</span>

                <span class="c1">// copy data
</span>                <span class="n">tb_memcpy</span><span class="p">(</span><span class="n">data</span> <span class="o">+</span> <span class="n">read</span><span class="p">,</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">data</span> <span class="o">+</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span><span class="p">,</span> <span class="n">need</span><span class="p">);</span>

                <span class="c1">// update the read size for buffer
</span>                <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span> <span class="o">+=</span> <span class="n">need</span><span class="p">;</span>

                <span class="c1">// update the total read size
</span>                <span class="n">read</span> <span class="o">+=</span> <span class="n">need</span><span class="p">;</span>
            <span class="p">}</span>

            <span class="c1">// 将读空的buffer释放掉
</span>            <span class="k">if</span> <span class="p">(</span><span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span> <span class="o">==</span> <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">size</span><span class="p">)</span>
                <span class="n">tb_queue_pop</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">);</span>
        <span class="p">}</span>

        <span class="c1">// ok?
</span>        <span class="k">return</span> <span class="n">read</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="c1">// 实现wait回调接口，用于等待数据，tb_stream_wait/tb_stream_bread等阻塞读取接口会用到
</span>    <span class="k">static</span> <span class="n">tb_long_t</span> <span class="nf">tb_stream_real_wait</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">,</span> <span class="n">tb_size_t</span> <span class="n">wait</span><span class="p">,</span> <span class="n">tb_long_t</span> <span class="n">timeout</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return_val</span><span class="p">(</span><span class="n">rstream</span> <span class="o">&amp;&amp;</span> <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">,</span> <span class="o">-</span><span class="mi">1</span><span class="p">);</span>

        <span class="c1">// 当前是否有数据可读？
</span>        <span class="k">return</span> <span class="n">tb_queue_size</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">)</span><span class="o">?</span> <span class="n">TB_STREAM_WAIT_READ</span> <span class="o">:</span> <span class="n">TB_STREAM_WAIT_NONE</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="c1">// 实现ctrl回调接口，用于设置和获取一些状态，扩展一些自定义的接口，tb_stream_ctrl接口会用到
</span>    <span class="k">static</span> <span class="n">tb_bool_t</span> <span class="nf">tb_stream_real_ctrl</span><span class="p">(</span><span class="n">tb_stream_ref_t</span> <span class="n">stream</span><span class="p">,</span> <span class="n">tb_size_t</span> <span class="n">ctrl</span><span class="p">,</span> <span class="n">tb_va_list_t</span> <span class="n">args</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_stream_real_ref_t</span> <span class="n">rstream</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_stream_real_ref_t</span><span class="p">)</span><span class="n">stream</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return_val</span><span class="p">(</span><span class="n">rstream</span><span class="p">,</span> <span class="n">tb_false</span><span class="p">);</span>

        <span class="c1">// ctrl
</span>        <span class="k">switch</span> <span class="p">(</span><span class="n">ctrl</span><span class="p">)</span>
        <span class="p">{</span>
        <span class="k">case</span> <span class="n">TB_STREAM_CTRL_GET_SIZE</span><span class="p">:</span>
            <span class="p">{</span>
                <span class="c1">// the psize
</span>                <span class="n">tb_hong_t</span><span class="o">*</span> <span class="n">psize</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_hong_t</span><span class="o">*</span><span class="p">)</span><span class="n">tb_va_arg</span><span class="p">(</span><span class="n">args</span><span class="p">,</span> <span class="n">tb_hong_t</span><span class="o">*</span><span class="p">);</span>
                <span class="n">tb_assert_and_check_break</span><span class="p">(</span><span class="n">psize</span><span class="p">);</span>

                <span class="c1">// 获取数据流大小，tb_stream_size有用到
</span>                <span class="o">*</span><span class="n">psize</span> <span class="o">=</span> <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">size</span><span class="p">;</span>

                <span class="c1">// ok
</span>                <span class="k">return</span> <span class="n">tb_true</span><span class="p">;</span>
            <span class="p">}</span>   
            <span class="c1">// 在另外一端通过tb_stream_ctrl来不断的送入数据块到stream
</span>        <span class="k">case</span> <span class="n">TB_STREAM_CTRL_REAL_PUSH</span><span class="p">:</span>
            <span class="p">{</span>
                <span class="c1">// check
</span>                <span class="n">tb_assert_and_check_break</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">);</span>

                <span class="c1">// the data and size
</span>                <span class="n">tb_byte_t</span> <span class="k">const</span><span class="o">*</span>    <span class="n">data</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_byte_t</span> <span class="k">const</span><span class="o">*</span><span class="p">)</span><span class="n">tb_va_arg</span><span class="p">(</span><span class="n">args</span><span class="p">,</span> <span class="n">tb_byte_t</span> <span class="k">const</span><span class="o">*</span><span class="p">);</span>
                <span class="n">tb_size_t</span>           <span class="n">size</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_size_t</span><span class="p">)</span><span class="n">tb_va_arg</span><span class="p">(</span><span class="n">args</span><span class="p">,</span> <span class="n">tb_size_t</span><span class="p">);</span>
                <span class="n">tb_assert_and_check_break</span><span class="p">(</span><span class="n">data</span> <span class="o">&amp;&amp;</span> <span class="n">size</span><span class="p">);</span>

                <span class="c1">// 压入一个数据块
</span>                <span class="n">tb_real_buffer_t</span> <span class="n">buffer</span><span class="p">;</span>
                <span class="n">buffer</span><span class="p">.</span><span class="n">data</span> <span class="o">=</span> <span class="n">tb_memdup</span><span class="p">(</span><span class="n">data</span><span class="p">,</span> <span class="n">size</span><span class="p">);</span>
                <span class="n">buffer</span><span class="p">.</span><span class="n">size</span> <span class="o">=</span> <span class="n">size</span><span class="p">;</span>
                <span class="n">buffer</span><span class="p">.</span><span class="n">read</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
                <span class="n">tb_queue_put</span><span class="p">(</span><span class="n">rstream</span><span class="o">-&gt;</span><span class="n">buffers</span><span class="p">,</span> <span class="o">&amp;</span><span class="n">buffer</span><span class="p">);</span>

                <span class="c1">// 更新总的数据大小
</span>                <span class="n">rstream</span><span class="o">-&gt;</span><span class="n">size</span> <span class="o">+=</span> <span class="n">size</span><span class="p">;</span>

                <span class="c1">// ok
</span>                <span class="k">return</span> <span class="n">tb_true</span><span class="p">;</span>
            <span class="p">}</span>
        <span class="nl">default:</span>
            <span class="k">break</span><span class="p">;</span>
        <span class="p">}</span>

        <span class="c1">// failed
</span>        <span class="k">return</span> <span class="n">tb_false</span><span class="p">;</span>
    <span class="p">}</span>
</code></pre>
</div>

<p>通过上面四步， 基本上一个自定义流就实现好了，上面说的<code class="highlighter-rouge">tb_real_buffer_exit</code>主要用于queue维护的buffer的自动释放
详细说明和使用见容器章节，下面附属相关实现：</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="k">static</span> <span class="n">tb_void_t</span> <span class="nf">tb_real_buffer_exit</span><span class="p">(</span><span class="n">tb_element_ref_t</span> <span class="n">element</span><span class="p">,</span> <span class="n">tb_pointer_t</span> <span class="n">buff</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// check
</span>        <span class="n">tb_real_buffer_ref_t</span> <span class="n">buffer</span> <span class="o">=</span> <span class="p">(</span><span class="n">tb_real_buffer_ref_t</span><span class="p">)</span><span class="n">buff</span><span class="p">;</span>
        <span class="n">tb_assert_and_check_return</span><span class="p">(</span><span class="n">buffer</span><span class="p">);</span>

        <span class="c1">// exit it
</span>        <span class="k">if</span> <span class="p">(</span><span class="n">buffer</span><span class="o">-&gt;</span><span class="n">data</span><span class="p">)</span> <span class="n">tb_free</span><span class="p">(</span><span class="n">buffer</span><span class="o">-&gt;</span><span class="n">data</span><span class="p">);</span>
        <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">data</span> <span class="o">=</span> <span class="n">tb_null</span><span class="p">;</span>
        <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">size</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
        <span class="n">buffer</span><span class="o">-&gt;</span><span class="n">read</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
    <span class="p">}</span>
</code></pre>
</div>

<p>最后，贴下咱们这个自定义stream使用：</p>

<p>接收端</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// init stream
</span>    <span class="n">tb_stream_ref_t</span> <span class="n">stream</span> <span class="o">=</span> <span class="n">tb_stream_init_real</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">stream</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// open stream
</span>        <span class="k">if</span> <span class="p">(</span><span class="n">tb_stream_open</span><span class="p">(</span><span class="n">stream</span><span class="p">))</span>
        <span class="p">{</span>
            <span class="c1">// read line
</span>            <span class="n">tb_long_t</span> <span class="n">size</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
            <span class="n">tb_char_t</span> <span class="n">line</span><span class="p">[</span><span class="n">TB_STREAM_BLOCK_MAXN</span><span class="p">];</span>
            <span class="k">while</span> <span class="p">((</span><span class="n">size</span> <span class="o">=</span> <span class="n">tb_stream_bread_line</span><span class="p">(</span><span class="n">stream</span><span class="p">,</span> <span class="n">line</span><span class="p">,</span> <span class="k">sizeof</span><span class="p">(</span><span class="n">line</span><span class="p">)))</span> <span class="o">&gt;=</span> <span class="mi">0</span><span class="p">)</span>
            <span class="p">{</span>
                <span class="c1">// trace
</span>                <span class="n">tb_trace_i</span><span class="p">(</span><span class="s">"line: %s"</span><span class="p">,</span> <span class="n">line</span><span class="p">);</span>
            <span class="p">}</span>
        <span class="p">}</span>

        <span class="c1">// exit stream
</span>        <span class="n">tb_stream_exit</span><span class="p">(</span><span class="n">stream</span><span class="p">);</span>
    <span class="p">}</span>
</code></pre>
</div>

<p>基本上没什么变化，就是换了下stream的初始化创建接口</p>

<p>输入端</p>

<div class="language-c highlighter-rouge"><pre class="highlight"><code>    <span class="c1">// 将数据不停的送入stream中
</span>    <span class="k">while</span> <span class="p">(</span><span class="mi">1</span><span class="p">)</span>
    <span class="p">{</span>
        <span class="c1">// fill data
</span>        <span class="n">tb_byte_t</span> <span class="n">data</span><span class="p">[</span><span class="mi">8192</span><span class="p">];</span>
        <span class="n">tb_memset</span><span class="p">(</span><span class="n">data</span><span class="p">,</span> <span class="mh">0xff</span><span class="p">,</span> <span class="k">sizeof</span><span class="p">(</span><span class="n">data</span><span class="p">));</span>

        <span class="c1">// push data
</span>        <span class="n">tb_stream_ctrl</span><span class="p">(</span><span class="n">stream</span><span class="p">,</span> <span class="n">TB_STREAM_CTRL_REAL_PUSH</span><span class="p">,</span> <span class="n">data</span><span class="p">,</span> <span class="k">sizeof</span><span class="p">(</span><span class="n">data</span><span class="p">));</span>
    <span class="p">}</span>
</code></pre>
</div>

<p>上面介绍的实现和使用方式，只是个例子，方便理解tbox中stream的机制，具体实现和使用还是需要根据自己的实际需求做调整。</p>

<p>更详细的使用和扩展，可参考源代码来了解。。</p>


        </article>
        <hr>

        <!-- baidu ads -->
        

        <!-- reward -->
        <div style="text-align: center;">
            <button id="rewardButton" disable="enable" onclick="var qr = document.getElementById('QR'); if (qr.style.display === 'none') {qr.style.display='block';} else {qr.style.display='none'}">
              <span>赏</span>
            </button>
            <div id="QR" style="display: none;">
                <div id="wechat" style="display: inline-block">
                  <img id="wechat_qr" src="/static/img/weixin.png" alt="WeChat Pay"/>
                  <p>微信打赏</p>
                </div>
                <div id="alipay" style="display: inline-block">
                  <img id="alipay_qr" src="/static/img/alipay.png" alt="Alipay"/>
                  <p>支付宝打赏</p>
                </div>
            </div>
        </div>

        
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
            
                
            
        
            
            
                
            
        
            
            
                
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
            
            
                
            
        
            
        
        

        
        
            
        
            
            
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
            
        
            
        
            
            
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
            
        
            
        
            
            
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
            
        
            
        
            
            
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
            
        
            
        
            
            
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
                
                    
                
                    
                
                    
                
            
            
        
        

        <div class="post-recent">
    <div class="pre">

        

        

        
        
        

        

        

        
        
        <p><strong>上一篇</strong> <a href="/cn/2016/02/25/update-v1.5.2/">tbox v1.5.2 更新!</a></p>
        
    </div>

    <div class="nex">

        

        

        
        
        <p><strong>下一篇</strong> <a href="/cn/2016/06/09/api-import/">插件开发之import类库</a></p>
        
    </div>
</div>


        <h2 id="comments">评论</h2>
        






<div id="gitalk-container"></div>
<link rel="stylesheet" href="/css/gitalk.css">
<script src="/js/gitalk.min.js"></script>

<script>
const gitalk = new Gitalk({
  clientID: '73946dc1d9e2276ad0da',
  clientSecret: '12a3cb94361ba3ebc6ecb68cf80d592bfaa8106d',
  repo: 'tboox.github.io',
  owner: 'waruqi',
  admin: ['waruqi'],
  id: location.pathname,       
  language: 'zh-CN',
  distractionFreeMode: false  
})

gitalk.render('gitalk-container')
</script>





    </div>
    <button class="anchor"><i class="fa fa-anchor"></i></button>
    <div class="right">
        <div class="wrap">

            <!-- codefund ads -->
            

            <!-- Content -->
            <div class="side content">
                <div>
                    内容
                </div>
                <ul id="content-side" class="content-ul">
                    <li><a href="#english">English</a></li>
                    <li><a href="#similar_posts">相关文章</a></li>
                    <li><a href="#comments">评论</a></li>
                </ul>
            </div>


            <!-- baidu ads -->
            
            
            <br>
            <div class="side">
                <div>
                    <i class="fa fa-external-link"></i>
                    链接
                </div>
                <ul class="content-ul">
                  <li><a href="http://github.com/waruqi/tbox">tbox</a></li>
                  <li><a href="http://www.xmake.io">xmake</a></li>
                  <li><a href="https://github.com/waruqi">github</a></li>
                </ul>
            </div> 

            <!-- xmake courses -->
            <br>
            <div class="side">
                <div>
                    <i class="fa fa-external-link"></i>
                    xmake 入门课程
                </div>
                <a href="https://xmake.io/#/zh-cn/about/course" target="_blank">
                <img src="/static/img/xmake-course.png" alt="course" width="256" height="193">
                </a>
            </div>

            <!-- qqgroup -->
            <br>
            <div class="side">
                <div>
                    <i class="fa fa-external-link"></i>
                    技术交流群（QQ）
                </div>
                <img src="/static/img/qqgroup.png" alt="qqgroup" width="256" height="284">
            </div> 

            <!-- google ads -->
            

            <!-- baidu ads -->
            

        </div>
    </div>

    <!-- baidu ads -->
    
</div>
<script>
/**
 * target _blank
 */
(function() {
    var aTags = document.querySelectorAll('article a:not([id])')
    for (var i = 0; i < aTags.length; i++) {
        aTags[i].setAttribute('target', '_blank')
    }
}());
</script>
<script src="/js/pageContent.js " charset="utf-8"></script>



    <footer class="site-footer">
    <div class="wrapper">
        <p class="description">
             Copyright (c) 2016-2020 tboox.org 
        </p>
        <p class="contact">
            
            <a href="https://github.com/waruqi" title="GitHub"><i class="fa fa-github" aria-hidden="true"></i></a> 
             
            
            <a href="mailto:waruqi@gmail.com" title="email"><i class="fa fa-envelope-o" aria-hidden="true"></i></a> 
            
            
            <a href="https://twitter.com/waruqi" title="Twitter"><i class="fa fa-twitter" aria-hidden="true"></i></a> 
            
            <a href="/feed.xml" title="feed"><i class="fa fa-feed" aria-hidden="true"></i></a> 
        </p>
        <p class="power">
            <span>
                Site powered by <a href="https://jekyllrb.com/">Jekyll</a> & <a href="https://github.com/Gaohaoyang">HyG</a> & <a href="https://pages.github.com/">Github Pages</a>.
            </span>
        </p>
    </div>
</footer>

    <div class="back-to-top">
    <a href="#top" class="scroll">
        <i class="fa fa-arrow-up" aria-hidden="true"></i>
    </a>
</div>

    <script src=" /js/main.js " charset="utf-8"></script>
    <script src=" /js/scroll.min.js " charset="utf-8"></script>
  </body>

</html>
